package com.cloudlead.common.worker.web;

import com.cloudlead.common.lang.Configs;
import com.cloudlead.common.lang.serialize.JSON;
import com.cloudlead.common.lang.utils.ObjectId;
import com.cloudlead.common.worker.ApiRegistry;
import com.cloudlead.common.worker.desc.ActionDesc;
import com.cloudlead.common.worker.desc.ModuleDesc;
import com.cloudlead.common.worker.desc.datatype.MapData;
import com.cloudlead.common.worker.impl.DefaultApiRegistry;
import com.sun.enterprise.ee.cms.core.*;
import com.sun.enterprise.ee.cms.impl.client.*;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Created by dean on 2015/7/29.
 */
public class WorkerManager {
    private static final Logger LOGGER = LoggerFactory
            .getLogger(WorkerManager.class);

    public static final String START_STATUS_REPORT_COMPNENT = "startStatusReport";

    private static WorkerManager singleInstance;

    private final GroupManagementService gms;

    private WorkerManager() {
        gms = (GroupManagementService) GMSFactory.startGMSModule(Configs.getString("worker.group.instance") + "-"
                        + ObjectId.getGenMachineId(),
                Configs.getString("worker.group"),
                GroupManagementService.MemberType.CORE, new Properties());
    }

    public static synchronized WorkerManager getInstance() {
        if (null == singleInstance) {
            singleInstance = new WorkerManager();
        }
        return singleInstance;
    }

    public GroupManagementService getGMS() {
        return gms;
    }

    public void joinGroup(WorkerInfo workerInfo, ApiRegistry apiRegistry, boolean onlyWatch) throws GMSException {

        if (!onlyWatch) {
            gms.addActionFactory(new JoinedAndReadyNotificationActionFactoryImpl(notification -> {
                LOGGER.debug(notification.getMemberToken() + " join and ready.");
                if (!notification.getMemberToken().equals(gms.getInstanceName())) {
                    try {
                        gms.getGroupHandle().sendMessage(notification.getMemberToken(), START_STATUS_REPORT_COMPNENT, getWorkerInfoMessage(workerInfo, apiRegistry));
                    } catch (GMSException e) {
                        e.printStackTrace();
                        LOGGER.error("GMS send message error.", e);
                    }
                }
            }));
        }

        gms.addActionFactory(new PlannedShutdownActionFactoryImpl(notification -> {
            LOGGER.debug(notification.getMemberToken() + " shutdown.");
            WorkerRegistry.getInstance().removeWorker(notification.getMemberToken());
        }));
        gms.addActionFactory(new FailureNotificationActionFactoryImpl(notification -> {
            LOGGER.debug(notification.getMemberToken() + " Undetectable heartbeat.");
            WorkerRegistry.getInstance().disableWorker(notification.getMemberToken());
        }));
        gms.addActionFactory(new JoinNotificationActionFactoryImpl(notification -> {
            LOGGER.debug(notification.getMemberToken() + " Join.");
            WorkerRegistry.getInstance().enableWorker(notification.getMemberToken());
        }));
        gms.addActionFactory(new MessageActionFactoryImpl(notification -> {
            if (notification instanceof MessageSignal) {
                MessageSignal messageSignal = (MessageSignal) notification;
                try {
                    String message = new String(messageSignal.getMessage(), "utf8");
                    LOGGER.debug("receive message :" + message);
                    WorkerInfo workerStatus = JSON.toBean(message, WorkerInfo.class);
                    if (StringUtils.isBlank(workerStatus.getWorkerName())) {
                        workerStatus.setWorkerName(messageSignal.getMemberToken());
                    }
                    WorkerRegistry.getInstance().addWorker(messageSignal.getMemberToken(), workerStatus);
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            }
        }), START_STATUS_REPORT_COMPNENT);

        gms.join();
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        gms.reportJoinedAndReadyState();

        if (!onlyWatch) {
            byte[] workerStatusReportMessage = getWorkerInfoMessage(workerInfo, apiRegistry);
            gms.getGroupHandle().sendMessage(START_STATUS_REPORT_COMPNENT, workerStatusReportMessage);
        }

    }


    private byte[] getWorkerInfoMessage(WorkerInfo workerInfo, ApiRegistry apiRegistry) {
        Map<String, ModuleDesc> allModuleDesc = apiRegistry.getAllModuleDesc();
        Map<String, Set<String>> modules = new HashMap<>();
        for (ModuleDesc moduleDesc : allModuleDesc.values()) {
            if (null != moduleDesc.getModuleFacades() && moduleDesc.getModuleFacades().size() > 0) {
                modules.put(moduleDesc.getFullName(), moduleDesc.getModuleFacades());
            }
        }
        workerInfo.setModules(modules);

        Set<String> moduleNames = modules.keySet();

        Map<String, ActionDesc> allApiDesc = apiRegistry.getAllApiDesc();
        Map<String, List<String>> actions = new HashMap<>();
        for (ActionDesc actionDesc : allApiDesc.values()) {
            if (moduleNames.contains(actionDesc.getModule().getFullName())) {
                MapData request = actionDesc.getRequest();
                MapData contentMap = (MapData) request.getMapping().get(DefaultApiRegistry.REQUEST_CONTENT_PARAM);
                List<String> paramNames = new ArrayList<>();
                paramNames.addAll(contentMap.getMapping().keySet());
                actions.put(actionDesc.getName(), paramNames);
            }
        }
        workerInfo.setActions(actions);

        return JSON.toString(workerInfo).getBytes();
    }

    public void leaveGroupAndShutdown() {
        gms.shutdown(GMSConstants.shutdownType.INSTANCE_SHUTDOWN);
    }


}
